package com.colin.util.util;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 自定义线程池
 * 分装了通用线程提交方法putThread
 * 分装了批量分页执行的方法dealAndGetAsync
 * 可使用自定义执行器ThreadPoolExecutor
 * @author zhaoliangze
 */
public class CustomThreadPool {
    private static final Logger logger = LoggerFactory.getLogger(CustomThreadPool.class);

    /**
     * 线程池执行器
     */
    private ThreadPoolExecutor executor;

    public CustomThreadPool(ThreadPoolExecutor executor) {
        this.executor = executor;
    }

    /**
     * 无返回值线程执行
     */
    public void putThread(Runnable runnable) {
        executor.submit(runnable);
    }

    /**
     * 定义一个多线程返回实体
     */
    public static class ThreadResult<T> implements Serializable {

        private static final long serialVersionUID = 7164771110480210650L;
        /**
         * 线程返回消息，异常消息返回
         */
        private String message;
        /**
         * 线程返回数据
         */
        private T data;

        public String getMessage() {
            return message;
        }

        public void setMessage(String message) {
            this.message = message;
        }

        public T getData() {
            return data;
        }

        public void setData(T data) {
            this.data = data;
        }
    }


    /**
     * 异步运行函数
     *
     * @param <T> 返回实体定义
     * @param <V> 入惨实体定义
     */
    @FunctionalInterface
    public interface RunFunction<T, V> {
        ThreadResult<T> run(List<V> list);
    }

    /**
     * 使用CompletableFuture方式获取线程返回值
     * 性能比Callable更优
     */
    private <T, V> CompletableFuture<ThreadResult<T>> putThreadWithReturn(List<V> list, RunFunction<T, V> runFunction) {
        return CompletableFuture.supplyAsync(() -> runFunction.run(list), executor);
    }

    /**
     * 异步处理集合任务并获取返回值
     *
     * @param list         待处理集合
     * @param pageSize     分页开启线程，若为10标识每10个元素开启一个线程处理
     * @param runFunctions 待处理业务逻辑
     */
    public <T, V> List<ThreadResult<T>> dealAndGetAsync(List<V> list, Integer pageSize, RunFunction<T, V> runFunctions) {
        List<ThreadResult<T>> results = new ArrayList<>();
        if (list == null || list.size() < 1) {
            return results;
        }

        if (list.size() < pageSize) {
            CompletableFuture<ThreadResult<T>> listCompletableFuture = putThreadWithReturn(list, runFunctions);
            results.add(listCompletableFuture.join());
            return results;
        }

        List<CompletableFuture<ThreadResult<T>>> completableFutures = new ArrayList<>();
        int total = list.size();
        int page = total / pageSize + 1;
        for (int i = 0; i < page; i++) {
            int start = i * pageSize;
            int end = Math.min((start + pageSize), total);
            List<V> longs = list.subList(start, end);
            if (CollectionUtils.isNotEmpty(longs)) {
                completableFutures.add(putThreadWithReturn(longs, runFunctions));
            }
        }

        completableFutures.forEach(future -> results.add(future.join()));

        return results;
    }

}
